#include "server.h"

static int CONN_TIMEOUT=10;

static struct worker_ctl
{/*{{{*/
    unsigned long thrd_id;
    int worker_idx;
    SOCKET listen_sock;
    char *request;
    char *respond;

    unsigned int seq;
    unsigned int *svc_call_count;
    unsigned int client_conn_count;
}/*}}}*/
*a_worker_ctl;

struct _svctrack
{/*{{{*/
    struct _svc
    {
        char name[NAME_SIZE+1];
        struct timeval tv_begin, tv_end;
    } *svcs;
    int svc_alloc_num;
    int svc_num;
    struct timeval tv_begin, tv_end;
};/*}}}*/

static void _svctrack_init(struct _svctrack *svctrack)
{/*{{{*/
    svctrack->svcs=(struct _svc *)NULL;
    svctrack->svc_alloc_num=svctrack->svc_num=0;
    gettimeofday(&svctrack->tv_begin, NULL);
    svctrack->tv_end.tv_sec=svctrack->tv_end.tv_usec=0;
}/*}}}*/

static int _svctrack_begin(struct _svctrack *svctrack, char *svc)
{/*{{{*/
    static int ALLOC_NUM=5;
    struct _svc *svcp;

    if(svctrack->svc_num == svctrack->svc_alloc_num)
    {
        EXPAND_ARRAY_DOERR(svctrack->svcs, struct _svc,
                ALLOC_NUM, svctrack->svc_alloc_num,
                errlog("%s: expand svctrack.svcs fail", __func__);
                return(-1));
    }
    svcp=&svctrack->svcs[svctrack->svc_num];
    ++svctrack->svc_num;
    strcpy(svcp->name, svc);
    gettimeofday(&svcp->tv_begin, NULL);
    svcp->tv_end.tv_sec=svcp->tv_end.tv_usec=0;
    return(0);
}/*}}}*/

static void _svctrack_end(struct _svctrack *svctrack)
{/*{{{*/
    gettimeofday(&svctrack->svcs[svctrack->svc_num-1].tv_end, NULL);
}/*}}}*/

static void _svctrack_dump(struct worker_ctl *worker_ctl,
        struct _svctrack *svctrack, unsigned int abort)
{/*{{{*/
    if(abort)
    {
        char *buf=worker_ctl->respond;
        char ts_begin[TIMESTAMP_SIZE+1];
        char ts_end[TIMESTAMP_SIZE+1];
        int offset=0;
        int i;

        gettimeofday(&svctrack->tv_end, NULL);
        timestamp_to_str(svctrack->tv_begin.tv_sec, ts_begin);
        offset+=sprintf(buf+offset,
                "%s: worker(%d) sess(%d) use %d millisecs, begin(%s)->",
                __func__,
                worker_ctl->worker_idx, worker_ctl->seq,
                diff_millisecond(svctrack->tv_begin, svctrack->tv_end),
                ts_begin+DATE_SIZE);
        for(i=0; i<svctrack->svc_num; i++)
        {
            struct _svc *svcp=&svctrack->svcs[i];
            timestamp_to_str(svcp->tv_begin.tv_sec, ts_begin);
            timestamp_to_str(svcp->tv_end.tv_sec, ts_end);
            offset+=sprintf(buf+offset, "%s(use:%d %s-%s)->",
                    svcp->name,
                    diff_millisecond(svcp->tv_begin, svcp->tv_end),
                    ts_begin+DATE_SIZE, ts_end+DATE_SIZE);
        }
        timestamp_to_str(svctrack->tv_end.tv_sec, ts_end);
        sprintf(buf+offset, "end(%s)", ts_end+DATE_SIZE);
        errlog(buf);
    }
    FREE(svctrack->svcs);
}/*}}}*/

static void _sess_disconn(struct worker_ctl *worker_ctl, SOCKET new_sock)
{/*{{{*/
    ++worker_ctl->client_conn_count;
    sock_close(new_sock);
}/*}}}*/

static int _sess_read(struct worker_ctl *worker_ctl, SOCKET new_sock)
{/*{{{*/
    char *request=worker_ctl->request;
    int request_size;
    int res;

    res=sock_read(new_sock, (unsigned char *)request, &request_size,
            CONN_TIMEOUT, WRAP);
    if(res)
    {
        int offset=0;
        char buf[BUFFER_SIZE];

        offset+=sprintf(buf, "%s: worker(%d:%d) peer ",
                __func__, worker_ctl->worker_idx, worker_ctl->seq);
        if(res == RET_ERR_ABORT)
            offset+=sprintf(buf+offset, "abort");
        else if(res == RET_ERR_TIMEOUT)
            offset+=sprintf(buf+offset, "timeout");
        else
            offset+=sprintf(buf+offset, "read fail, res:%d", res);
        errlog(buf);
        return(-1);
    }
    request[request_size]=ZERO;
    return(0);
}/*}}}*/

static int _sess_write(struct worker_ctl *worker_ctl, SOCKET new_sock)
{/*{{{*/
    char *respond=worker_ctl->respond;
    int respond_size;
    int res;

    respond_size=strlen(respond);
    res=sock_write(new_sock, (unsigned char *)respond, &respond_size,
            CONN_TIMEOUT, WRAP);
    if(res)
    {
        errlog("%s: worker(%d:%d) respond write fail, res:%d",
                __func__,
                worker_ctl->worker_idx, worker_ctl->seq, res);
        return(-1);
    }
    return(0);
}/*}}}*/

static int _call_svc(struct worker_ctl *worker_ctl,
        struct _svctrack *svctrack, char *svc, char *request)
{/*{{{*/
    int worker_idx=worker_ctl->worker_idx;
    char *respond=worker_ctl->respond;
    int rtn;

    struct svc_map *map=NULL;
    int svc_idx;
    for(svc_idx=0; svc_idx<SVC_MAP_SIZE; ++svc_idx)
    {
        map=&SVC_MAP[svc_idx];
        if(strcasecmp(map->name, svc) == 0) break;
    }
    if(svc_idx == SVC_MAP_SIZE)
    {
        errlog("%s: worker(%d) proc_(%s) not found",
                __func__, worker_idx, svc);
        para_put_item(respond, SOAR_ERR_SVC_NOT_FOUND);
        rtn=atoi(SOAR_ERR_SVC_NOT_FOUND);
    }
    else
    {
        _svctrack_begin(svctrack, svc);
        rtn=map->pfunc(worker_idx, request, respond);
        _svctrack_end(svctrack);
        ++worker_ctl->svc_call_count[svc_idx];
    }

    return(rtn);
}/*}}}*/

static int _proc(struct worker_ctl *worker_ctl, SOCKET new_sock)
{/*{{{*/
    int worker_idx=worker_ctl->worker_idx;
    char *request=worker_ctl->request;
    char *respond=worker_ctl->respond;
    unsigned int end=IS_FALSE;
    unsigned int abort=IS_FALSE;
    struct _svctrack svctrack;
    char *svc;
    char *body;
    int res;

    if(++worker_ctl->seq == 10000)
        worker_ctl->seq=0;

    _svctrack_init(&svctrack);

    while(!end)
    {
        // 读取
        res=_sess_read(worker_ctl, new_sock);
        if(res)
        {
            abort=IS_TRUE;
            break;
        }
#ifdef _MY_DEBUG
        errlog("%s: _sess_read", __func__);
#endif
    
        // 解析服务
        CLR_BUF(respond);
        body=NULL;
        svc=para_get_item(request, &body);
        if(!svc || (res=strlen(svc)) <= 0 || res > NAME_SIZE)
        {
            errlog("%s: worker(%d) svc(%s) invalid",
                    __func__, worker_idx, (svc)?svc:SPACES);
            para_put_item(respond, SOAR_ERR_SVC_INVALID);
            goto RESPOND;
        }

        if(strcasecmp(svc, "CLOSE") == 0 ||
                strcasecmp(svc, "COMMIT") == 0 ||
                strcasecmp(svc, "ABORT") == 0)
        {
            para_put_item(respond, SOAR_ERR_OK);
            end=IS_TRUE;
#ifdef _MY_DEBUG
            errlog("%s: %s", __func__, svc);
#endif
        }
        else
        {
            _call_svc(worker_ctl, &svctrack, svc, body);
        }

RESPOND:
        // 返回
        res=_sess_write(worker_ctl, new_sock);
        if(res)
        {
            abort=IS_TRUE;
            break;
        }
    }

    _svctrack_dump(worker_ctl, &svctrack, abort);

    _sess_disconn(worker_ctl, new_sock);

    return(0);
}/*}}}*/

static void _worker(struct worker_ctl *worker_ctl)
{/*{{{*/
    static int TIMEOUT=1;
    static int UPDATE_STAT_INFO_INTERVAL=10;
    static int CHECK_OPERATE_INFO_INTERVAL=1;

    int worker_idx=worker_ctl->worker_idx;
    unsigned int quit=IS_FALSE;
    time_t last_stat=0;
    time_t last_operate=0;
    time_t now;

    while(!quit)
    {
        SOCKET new_sock=sock_accept(worker_ctl->listen_sock, TIMEOUT);
        if(new_sock < 0)
        {
            if(new_sock != RET_ERR_TIMEOUT)
            {
                errlog("%s: sock_accept fail, err:%d",
                        __func__, get_last_error(NULL));
            }
        }
        else
        {
            _proc(worker_ctl, new_sock);
        }

        now=curr_timestamp(NULL);

        // update client & svc info in stat_info
        if(now - last_stat > UPDATE_STAT_INFO_INTERVAL)
        {/*{{{*/
            int call_count=0;

            if(worker_ctl->client_conn_count > 0)
            {
                pthread_mutex_lock(&stat_info._svc_.lock);
                int i;
                for(i=0; i<SVC_MAP_SIZE; ++i)
                {
                    if(worker_ctl->svc_call_count[i] > 0)
                    {
                        struct _svc_info *svc=&stat_info._svc_.svc[i];
                        svc->call_num+=worker_ctl->svc_call_count[i];
                        call_count+=worker_ctl->svc_call_count[i];
                        worker_ctl->svc_call_count[i]=0;
                    }
                }
                pthread_mutex_unlock(&stat_info._svc_.lock);

                pthread_mutex_lock(&stat_info._worker_.lock);
                struct _worker_info *worker=
                    &stat_info._worker_.worker[worker_idx];
                worker->client_conn_num+=worker_ctl->client_conn_count;
                worker->client_call_num+=call_count;
                pthread_mutex_unlock(&stat_info._worker_.lock);
                worker_ctl->client_conn_count=0;
            }

            last_stat=now;
        }/*}}}*/

        // check operate_info
        if(now - last_operate > CHECK_OPERATE_INFO_INTERVAL)
        {/*{{{*/
            pthread_mutex_lock(&operate_info._worker_.lock);
            if((quit=operate_info._worker_.quit))
            {
                operate_info._worker_.state[worker_idx]=THRD_SHUTDOWN;
            }
            pthread_mutex_unlock(&operate_info._worker_.lock);

            last_operate=now;
        }/*}}}*/
    }

    errlog("%s: worker(%d) shutdown", __func__, worker_idx);
}/*}}}*/

int worker_start()
{/*{{{*/
    // stat_info
    pthread_mutex_init(&stat_info._worker_.lock, NULL);
    ALLOC_MULTI_DOERR(stat_info._worker_.worker, struct _worker_info,
            cnfg_info.workers,
            fprintf(stderr, "%s: alloc stat_info._worker_.worker fail\n",
                __func__);
            return(-1));
    int worker_idx;
    for(worker_idx=0; worker_idx<cnfg_info.workers; ++worker_idx)
    {
        struct _worker_info *worker=
            &stat_info._worker_.worker[worker_idx];
        worker->client_conn_num=0;
        worker->client_call_num=0;
    }

    pthread_mutex_init(&stat_info._svc_.lock, NULL);
    ALLOC_MULTI_DOERR(stat_info._svc_.svc, struct _svc_info,
            SVC_MAP_SIZE,
            fprintf(stderr, "%s: alloc stat_info._svc_.svc\n",
                __func__);
            return(-1));
    int i;
    for(i=0; i<SVC_MAP_SIZE; ++i)
    {
        struct _svc_info *svc=&stat_info._svc_.svc[i];
        svc->name=SVC_MAP[i].name;
        svc->call_num=0;
    }

    // operate_info
    pthread_mutex_init(&operate_info._worker_.lock, NULL);
    operate_info._worker_.quit=IS_FALSE;
    ALLOC_MULTI_DOERR(operate_info._worker_.state, enum thrd_state,
            cnfg_info.workers,
            fprintf(stderr,
                "%s: alloc operate_info._worker_.state fail\n",
                __func__);
            return(-1));
    for(worker_idx=0; worker_idx<cnfg_info.workers; ++worker_idx)
    {
        operate_info._worker_.state[worker_idx]=THRD_RUNNING;
    }

    // server port
    SOCKET listen_sock=sock_listen(cnfg_info._access_.server_port);
    if(listen_sock == INVALID_SOCKET)
    {
        fprintf(stderr, "%s: open server port fail\n", __func__);
        return(-1);
    }
    fprintf(stderr, "%s: open server port(%d)\n",
            __func__, cnfg_info._access_.server_port);

    // start
    ALLOC_MULTI_DOERR(a_worker_ctl, struct worker_ctl,
            cnfg_info.workers,
            fprintf(stderr, "%s: alloc a_worker_ctl fail\n", __func__);
            return(-1));
    for(worker_idx=0; worker_idx<cnfg_info.workers; ++worker_idx)
    {
        struct worker_ctl *worker_ctl=&a_worker_ctl[worker_idx];

        worker_ctl->listen_sock=listen_sock;
        worker_ctl->worker_idx=worker_idx;
        ALLOC_MULTI_DOERR(worker_ctl->request, char, SOCK_MSG_SIZE,
                fprintf(stderr, "%s: alloc worker(%d) request fail\n",
                    __func__, worker_idx);
                return(-1);
                );
        ALLOC_MULTI_DOERR(worker_ctl->respond, char, SOCK_MSG_SIZE,
                fprintf(stderr, "%s: alloc worker(%d) respond fail\n",
                    __func__, worker_idx);
                return(-1);
                );
        worker_ctl->seq=0;
        ALLOC_MULTI_DOERR(worker_ctl->svc_call_count, unsigned int,
                SVC_MAP_SIZE,
                fprintf(stderr, "%s: alloc worker(%d) svc_count fail\n",
                    __func__, worker_idx);
                return(-1));
        for(i=0; i<SVC_MAP_SIZE; ++i)
            worker_ctl->svc_call_count[i]=0;
        worker_ctl->client_conn_count=0;

        if(thrd_create((THRD_FUNC_T)_worker, (void *)worker_ctl,
                    THRD_DETACH, &worker_ctl->thrd_id))
        {
            fprintf(stderr, "%s: worker(%d) thrd create fail\n",
                    __func__, worker_idx);
            return(-1);
        }
    }

    return(0);
}/*}}}*/
